package com.doit.shark.dsp.pre

import ch.hsr.geohash.GeoHash
import com.doit.shark.dsp.beans.{DspLogBean, DspLogBeanStr}
import com.doit.shark.dsp.utils.GenLogBean
import org.apache.commons.lang3.StringUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{Dataset, SparkSession}
import redis.clients.jedis.Jedis

/**
  * Created by hunter.coder 涛哥  
  * 2019/4/8 18:12
  * 交流qq:657270652
  * Version: 1.0
  * 更多学习资料：https://blog.csdn.net/coderblack/
  * Description:  DSP广告竞价请求日志预处理程序
  * 过滤掉标识字段全为空的记录
  * 集成商圈信息、app信息
  * 输出为parquet文件
  *
  **/
object PreProcess {


  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.WARN)

    val spark = SparkSession
      .builder()
      .appName(PreProcess.getClass.getSimpleName)
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._

    // 加载原始日志
    val ds1: Dataset[String] = spark.read.textFile("g:/sharkdata/dsplog")

    // 清洗过滤
    val ds2: Dataset[DspLogBeanStr] = ds1.map(line => {
      val fields = line.split(",", -1)
      // 将数据封装成样例类
      GenLogBean.arr2LogBeanStr(fields)
    }).filter(bean => {
      StringUtils.isNotBlank(
        bean.imei + bean.imeimd5 + bean.imeisha1
          + bean.idfa + bean.idfamd5 + bean.idfasha1
          + bean.mac + bean.macmd5 + bean.macsha1
          + bean.androidid + bean.androididmd5 + bean.androididsha1
          + bean.openudid + bean.openudidmd5 + bean.openudidsha1)
    })

    // 数据集成
    val ds3 = ds2.mapPartitions(part => {

      // 建redis连接
      val jedis1 = new Jedis("c701", 6379)
      jedis1.select(1) // 省市区库

      val jedis2 = new Jedis("c701", 6379)
      jedis2.select(2) // 商圈名称库


      val jedis3 = new Jedis("c701", 6379)
      jedis3.select(3) // app信息库

      val res = part.map(bean => {

        // 取出经纬度
        val lng = bean.lng.toDouble
        val lat = bean.lat.toDouble


        if (lng >= -180 && lng <= 180 && lat >= -90 && lat <= 90) {
          // 转成geohashcode
          val geohashcode = GeoHash.withCharacterPrecision(lat, lng, 6).toBase32

          // 去redis中查询省市区 -- 日志中已经加进去了
          /* val area = jedis1.get(geohashcode.substring(0,geohashcode.length-1)).split(",") */

          // 去redis中查询商圈信息
          val bizName = jedis2.get(geohashcode)
          bean.bizName = bizName
        }

        // 去redis中查询app信息
        val appInfo = jedis3.get(bean.appid)
        var appDesc = ""

        if (appInfo != null) {
          val appName_Desc = appInfo.split("\001")
          bean.appname = appName_Desc(0)
          bean.appDesc = appName_Desc(1)
        }

        bean
      })

      jedis1.close()
      jedis2.close()
      jedis3.close()

      // 返回结果
      res
    })


    // 将结果保存为parquet文件
    ds3.toDF().write.parquet("g:/sharkdata/dsppreout/")


    spark.close()


  }

}
